package org.spark.core.transformations.java;

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.spark.sparkUtil.SparkJavaContextUtil;

import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

/**
 * 功能概述: flatMap案例：将文本行拆分为多个单词
 * Datetime:    2020/5/21   11:55
 * Author:   某人的目光
 */
public class FlatMap {
    public static void main(String[] args) {
        JavaSparkContext sc = SparkJavaContextUtil.getSparkStart("FlatMap");
        // 构造集合
        List<String> lineList = Arrays.asList("hello you", "hello me", "hello world");
        // 并行化集合，创建RDD
        JavaRDD<String> lines = sc.parallelize(lineList);
        // 对RDD执行flatMap算子，将每一行文本，拆分为多个单词
        // flatMap算子，在java中，接收的参数是FlatMapFunction
        // 我们需要自己定义FlatMapFunction的第二个泛型类型，即，代表了返回的新元素的类型
        // call()方法，返回的类型，不是U，而是Iterable<U>，这里的U也与第二个泛型类型相同
        // flatMap其实就是，接收原始RDD中的每个元素，并进行各种逻辑的计算和处理，返回可以返回多个元素
        // 多个元素，即封装在Iterable集合中，可以使用ArrayList等集合
        // 新的RDD中，即封装了所有的新元素；也就是说，新的RDD的大小一定是 >= 原始RDD的大小
        JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
            // 在这里会，比如，传入第一行，hello you
            // 返回的是一个Iterable<String>(hello, you)
            public Iterator<String> call(String str) throws Exception {
                return Arrays.asList(str.split(" ")).iterator();
            }
        });
        // 打印新的RDD
        words.foreach(new VoidFunction<String>() {


            public void call(String t) throws Exception {
                System.out.println(t);
            }
        });

        // 关闭JavaSparkContext
        sc.close();
    }
}
